package org.apache.solr.ocean.core;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

import org.apache.lucene.analysis.Analyzer;
import org.apache.lucene.store.RAMDirectory;
import org.apache.solr.ocean.Documents;
import org.apache.solr.ocean.core.Batch.MasterBatch;
import org.apache.solr.ocean.core.Index.DeletesResult;
import org.apache.solr.ocean.core.Index.IndexSnapshot;
import org.apache.solr.ocean.replication.Replication;
import org.apache.solr.schema.IndexSchema;

public class Transaction {
	private Batch batch;
	private CountDownLatch latch;
	private List<Failure> failures = new ArrayList<Failure>();
	private List<DeletesResult> deletesResults = new ArrayList<DeletesResult>();
	private List<IndexSnapshot> newIndexSnapshots = new ArrayList<IndexSnapshot>();
	private ReentrantLock lock = new ReentrantLock();
	private Condition goCondition = lock.newCondition();
	private Long id;
	private Long previousId;
	private Replication replication;
	private CommitResult commitResult;
	private TransactionSystem.Category category;
	private IndexSchema schema;

	// TODO: add timeout to transaction
	public Transaction(Long id, Long previousId, Batch batch, WriteableMemoryIndex writeableIndex, List<Index> nonWriteableIndices, ExecutorService commitThreadPool, TransactionSystem.Category category, IndexSchema schema) throws Exception {
		this.id = id;
		this.previousId = previousId;
		this.batch = batch;
		this.category = category;
		this.schema = schema;
		this.replication = category.getTransactionSystem().getReplication();
		List<Callable<DeletesResult>> tasks = new ArrayList<Callable<DeletesResult>>();
		Deletes deletes = batch.getDeletes();
		if (batch.hasDeletes()) {
			for (Index index : nonWriteableIndices) {
				tasks.add(new DeletesTask(deletes, index, this));
			}
		}
		int numDocsAdded = 0;
		if (batch.hasRAMDirectory()) {
			tasks.add(new AddRamIndexDocumentsTask(batch.getRamDirectory()));
		} else if (batch.hasDocuments()) {
			Documents documents = batch.getDocuments();
			Analyzer analyzer = batch.getAnalyzer();
			tasks.add(new AddWriteableMemoryDocumentsTask(documents, analyzer, deletes, writeableIndex));
			numDocsAdded += documents.size();
		} else {
			tasks.add(new DeletesTask(deletes, writeableIndex, this));
		}
		latch = new CountDownLatch(tasks.size());
		commitThreadPool.invokeAll(tasks);
		latch.await();
		lock.lock();
		try {
			goCondition.signalAll();
		} finally {
			lock.unlock();
		}
		if (failures.size() == 0) {
			commitResult = new CommitResult(id, deletesResults, numDocsAdded, writeableIndex.getId());
		} else {
			throw new Exception("transaction failed " + failures);
		}
	}

	public IndexSchema getIndexSchema() {
		return schema;
	}

	public List<IndexSnapshot> getNewIndexSnapshots() {
		return newIndexSnapshots;
	}

	public Long getPreviousId() {
		return previousId;
	}

	public CommitResult getCommitResult() {
		assert commitResult != null; // should have thrown exception before this
																	// point
		return commitResult;
	}

	public static class CommitResult {
		private Long snapshotID;
		private List<DeletesResult> deletesResults;
		private Integer numAdded;
		private Long addedIndexID;

		public CommitResult(Long snapshotID, List<DeletesResult> deletesResults, Integer numAdded, Long addedIndexID) {
			this.snapshotID = snapshotID;
			this.deletesResults = deletesResults;
			this.numAdded = numAdded;
			this.addedIndexID = addedIndexID;
		}

		public int getNumDocChanges() {
			int numChanged = 0;
			if (numAdded != null)
				numChanged += numAdded;
			if (deletesResults != null) {
				for (DeletesResult deletesResult : deletesResults) {
					numChanged += deletesResult.getNumDeleted();
				}
			}
			return numChanged;
		}

		public Long getSnapshotID() {
			return snapshotID;
		}

		public List<DeletesResult> getDeletesResults() {
			return deletesResults;
		}

		public Integer getNumAdded() {
			return numAdded;
		}

		public Long getAddedIndexID() {
			return addedIndexID;
		}
	}

	public Long getId() {
		return id;
	}

	public abstract static class Failure extends Exception {
		public Failure(Throwable throwable) {
			super(throwable);
		}
	}

	public static class LogFailure extends Failure {
		public LogFailure(Throwable throwable) {
			super(throwable);
		}
	}

	public static class IndexFailure extends Failure {
		Index index;

		public IndexFailure(Index index, Throwable throwable) {
			super(throwable);
			this.index = index;
		}
	}

	public static class DeletesTask implements Callable<DeletesResult> {
		private Index index;
		private Deletes deletes;
		private Transaction transaction;

		public DeletesTask(Deletes deletes, Index index, Transaction transaction) {
			this.deletes = deletes;
			this.index = index;
			this.transaction = transaction;
		}

		public DeletesResult call() throws Exception {
			DeletesResult deletesResult = index.commitDeletes(deletes, transaction);
			transaction.addDeletesResult(deletesResult);
			return deletesResult;
		}
	}

	public class AddRamIndexDocumentsTask implements Callable<DeletesResult> {
		private RAMDirectory ramDirectory;

		public AddRamIndexDocumentsTask(RAMDirectory ramDirectory) {
			this.ramDirectory = ramDirectory;
		}

		public DeletesResult call() throws Exception {
			// TODO: create new ramindex
			Long indexId = category.getNextIndexId();
			Analyzer analyzer = batch.getAnalyzer();
			RamIndex ramIndex = new RamIndex(indexId, id, null, ramDirectory, category, schema);
			IndexSnapshot indexSnapshot = ramIndex.commitIndex(Transaction.this);
			DeletesResult deletesResult = new DeletesResult(indexId);
			addDeletesResult(deletesResult);
			return deletesResult;
		}
	}

	public class AddWriteableMemoryDocumentsTask implements Callable<DeletesResult> {
		private Documents documents;
		private Analyzer analyzer;
		private Deletes deletes;
		private WriteableMemoryIndex writeableIndex;

		public AddWriteableMemoryDocumentsTask(Documents documents, Analyzer analyzer, Deletes deletes, WriteableMemoryIndex writeableIndex) {
			this.documents = documents;
			this.analyzer = analyzer;
			this.deletes = deletes;
			this.writeableIndex = writeableIndex;
		}

		public DeletesResult call() throws Exception {
			DeletesResult deletesResult = writeableIndex.commitChanges(documents, deletes, analyzer, Transaction.this);
			addDeletesResult(deletesResult);
			return deletesResult;
		}
	}

	void addDeletesResult(DeletesResult deletesResult) {
		deletesResults.add(deletesResult);
	}

	void failed(Index index, Throwable throwable) {
		failures.add(new IndexFailure(index, throwable));
		latch.countDown();
	}

	void ready(Index index) {
		latch.countDown();
	}

	public boolean go() throws InterruptedException {
		lock.lock();
		try {
			goCondition.await();
			if (failures.size() == 0) {
				try {
					if (batch instanceof MasterBatch) {
						MasterBatch masterBatch = (MasterBatch) batch;
						if (masterBatch.hasDeletes()) {
							int numDocIds = 0;
							for (DeletesResult deletesResult : deletesResults) {
								numDocIds += deletesResult.getDocIds().size();
							}
							long[] docIds = new long[numDocIds];
							int x = 0;
							for (DeletesResult deletesResult : deletesResults) {
								for (Long docId : deletesResult.getDocIds()) {
									docIds[x] = docId;
									x++;
								}
							}
							masterBatch.getDeletes().setDocIds(docIds);
						}
						replication.writeMasterBatch(id, previousId, masterBatch);
					}
				} catch (Throwable throwable) {
					failures.add(new LogFailure(throwable));
					return false;
				}
				return true;
			} else {
				return false;
			}
		} finally {
			lock.unlock();
		}
	}
}
